package com.zik.rader_and_video.client;

import com.zik.rader_and_video.config.MqttConfig;
import com.zik.rader_and_video.po.MqttRespData;
import com.zik.rader_and_video.server.WebSocketServer;
import com.zik.rader_and_video.utils.SleepAnalysisUtils;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;

/**
 * @author zik
 * @Time 2023/6/14 15:15
 * @description 消息回调类
 **/
@Slf4j
@Component
public class PushCallback implements MqttCallback {
    @Resource(name = "MqttConfig")
    private MqttConfig mqttConfig;

    /**
     * 线程池
     */
    @Resource(name = "threadPoolInstance")
    private ExecutorService executorService;

    @Override
    public void connectionLost(Throwable cause) {
        // 连接丢失后，一般在这里面进行重连
        log.info("连接断开，正在重连");
        mqttConfig.getMqttPushClient();
    }

    /**
     * 发送消息，消息到达后处理方法
     */
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.info("deliveryComplete---------" + token.isComplete());
    }

    /**
     * 订阅主题接收到消息处理方法
     * 多线程处理数据
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) {
        executorService.execute(() -> {
            // 判断是否为有用的数据
            if (SleepAnalysisUtils.isUseful(message)) {
                // 将数据封装为对象
                MqttRespData mqttRespData = SleepAnalysisUtils.getParam(message);
                System.out.println(mqttRespData);
                // 判断是否为非主动上报数据
                if (SleepAnalysisUtils.isRespData(mqttRespData)) {
                    log.info("线程{} -> 接受到数据 {}", Thread.currentThread().getName(),mqttRespData);
                    SleepAnalysisUtils.processMqttData(mqttRespData);
                }
            }

        });
    }

}
